package com.chenwy.example;
import akka.actor.ActorSystem;
import akka.dispatch.*;
import akka.japi.Function;
import akka.japi.Function2;
import scala.Tuple2;
import scala.concurrent.Await;
import scala.concurrent.ExecutionContextExecutorService;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import scala.util.Either;
import scala.util.Try;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class AkkaTest {

    public static void main(String[] args) {
        ExecutionContextExecutorService ec = ExecutionContexts.fromExecutorService(Executors.newCachedThreadPool());
        Future<String> f1 = Futures.successful("future1");
        Future<String> f2 = Futures.future(() -> { Thread.sleep(100); return "future2";}, ec);
        ActorSystem actor = ActorSystem.create("test");
        Future<String> f3 = Futures.future(() -> "future3", actor.dispatcher());
        Future f4 = akka.pattern.Patterns.ask(actor.actorSelection(""),"msg",1000 * 60);
        try {
            //阻塞式
//            System.out.println(Await.result(f1, Duration.create(1, TimeUnit.SECONDS)));
//            System.out.println(Await.result(f2, Duration.create(1, TimeUnit.SECONDS)));
//            System.out.println(Await.result(f3, Duration.create(1, TimeUnit.SECONDS)));

            //异步
//            f1.onComplete(new OnComplete<String>() {
//                @Override
//                public void onComplete(Throwable throwable, String s) throws Throwable {
//                    System.out.println(s);
//                }
//            },ec);
//
//            f1.onSuccess(new OnSuccess<String>() {
//                @Override
//                public void onSuccess(String s) throws Throwable {
//                    System.out.println(s);
//                }
//            },ec);
//
//            f1.onFailure(new OnFailure() {
//                @Override
//                public void onFailure(Throwable throwable) throws Throwable {
//                    throwable.printStackTrace();
//                }
//            },ec);

            //处理返回结果
//            Future<Integer> fi = f1.flatMap(new Mapper<String, Future<Integer>>() {
//                @Override
//                public Future<Integer> apply(String parameter) {
//                    return Futures.future(new Callable<Integer>() {
//                        @Override
//                        public Integer call() throws Exception {
//                            return parameter.length();
//                        }
//                    },ec);
//                }
//            },ec);

//            System.out.println(Await.result(fi, Duration.create(5, "s")));     //阻塞式

//            Future<String> fi = f1.flatMap(new Mapper<String, Future<String>>() {
//                @Override
//                public Future<String> apply(final String parameter1) {
//
//                    return f2.map(new Mapper<String, String>() {
//                        @Override
//                        public String apply(String parameter2) {
//                            return parameter1 +" "+parameter2;
//                        }
//                    },ec);
//                }
//            },ec);
//
//            System.out.println(Await.result(fi, Duration.create(5, "s")));     //阻塞式

//筛选，不匹配则抛异常
//            Future<String> fi = f2.filter(Filter.filterOf(new Function<String, Boolean>() {
//                @Override
//                public Boolean apply(String s) throws Exception {
//                    System.out.println();
//                    return s.length() == 7;
//                }
//            }),ec);
//            System.out.println(Await.result(fi, Duration.create(5, "s")));

//    组合Futures
//    sequence（将 T[Future[A]] 转换为 Future[T[A]]）
//            List<Future<String>> futureList = new ArrayList<Future<String>>();
//            futureList.add(f1);
//            futureList.add(f2);
//            futureList.add(f3);
//
//           Future<Iterable<String>> futures = Futures.sequence(futureList,ec);
//
//           Future<String> fi = futures.map(new Mapper<Iterable<String>, String>() {
//               @Override
//               public String apply(Iterable<String> parameter) {
//                  StringBuffer sb = new StringBuffer();
//                  for(String s : parameter){
//                      sb.append( s);
//                  }
//                  return sb.toString();
//               }
//           },ec);
//            System.out.println(Await.result(fi, Duration.create(5, "s")));

//     traverse （将 T[A] 转换为 Future[T[A]]）
//            List<String> list = Arrays.asList("Hello","akka");
//
//            Future<Iterable<String>> future = Futures.traverse(list, new Function<String, Future<String>>() {
//                @Override
//                public Future<String> apply(String s) throws Exception {
//                    return Futures.future(new Callable<String>() {
//                        @Override
//                        public String call() throws Exception {
//                            return s.toUpperCase();
//                        }
//                    },ec);
//                }
//            },ec);
//
//            Future<String> fi = future.map(new Mapper<Iterable<String>, String>() {
//                @Override
//                public String apply(Iterable<String> parameter) {
//                    StringBuffer sb = new StringBuffer();
//                    parameter.forEach(s -> {sb.append(s).append(" ");});
//                    return sb.toString();
//                }
//            },ec);
//            System.out.println(Await.result(fi, Duration.create(5, "s")));

//            fold（从一个初始值开始递归地对Future序列进行处理（它将sequence和map操作合并成一步了））
//            List<Future<String>> futureList = new ArrayList<Future<String>>();
//            futureList.add(f1);
//            futureList.add(f2);
//            futureList.add(f3);
//
//            Future<String> fi = Futures.fold("Init", futureList, new Function2<String, String, String>() {
//                @Override
//                public String apply(String s, String s2) throws Exception {
//                    return s + "+"+s2;
//                }
//            },ec);
//            System.out.println(Await.result(fi, Duration.create(5, "s")));

//            reduce（如果不想从给定的初始值开始递归，而想从future序列的第一个开始，则用reduce（它将sequence和map合并成一步了））
//            List<Future<String>> futureList = new ArrayList<Future<String>>();
//            futureList.add(f1);
//            futureList.add(f2);
//            futureList.add(f3);
//
//            Future<String> fi = Futures.reduce( futureList, new Function2<String, String, String>() {
//                @Override
//                public String apply(String s, String s2) throws Exception {
//                    return s + "+"+s2;
//                }
//            },ec);
//            System.out.println(Await.result(fi, Duration.create(7, "s")));

//            andThen（由于回调的执行是无序的，而且可能是并发执行的, 当你需要一组有序操作的时候需要一些技巧。）
//            Future<String> future = Futures.future(() -> "hello",ec).andThen(new OnComplete<String>() {
//                @Override
//                public void onComplete(Throwable throwable, String s) throws Throwable {
//                    System.out.println("先收到：" + s);
//                }
//            },ec).andThen(new OnComplete<String>() {
//                @Override
//                public void onComplete(Throwable failure, String success) {
//                    System.out.println("又收到：" + success);
//                }
//            },ec).andThen(new OnSuccess<Try<String>>() {
//                @Override
//                public void onSuccess(Try<String> stringTry) throws Throwable {
//                    System.out.println("收到onSuccess：" + stringTry);
//                }
//            }, ec);

//            fallbackTo（将两个 Futures 合并成一个新的 Future, 如果第一个Future失败了，它将持有第二个 Future 的成功值）
//            Future<String> f5 = Futures.failed(new RuntimeException("ex1"));
//            Future<String> f6 = Futures.failed(new RuntimeException("ex2"));
//            Future<String> f7 = Futures.successful("ok");
//            //fallbackTo 将两个 Futures 合并成一个新的 Future, 如果第一个Future失败了，它将持有第二个 Future 的成功值
//            Future<String> fr = f5.fallbackTo(f6).fallbackTo(f7);
//
//            System.out.println(Await.result(fr, Duration.create(5, "s")));

//            zip（操作将两个 Futures 组合压缩成一个新的Future，返回的新的Future持hold一个tuple实例，它包含二者成功的结果）
//            Future<String> f8 = Futures.future(new Callable<String>() {
//                @Override
//                public String call() throws Exception {
//                    System.out.println("future8" + Thread.currentThread().getName());
//                    Thread.sleep(1000 * 10);
//                    return "hello";
//                }
//            },ec);
//
//            Future<String> f9 = Futures.future(new Callable<String>() {
//                @Override
//                public String call() throws Exception {
//                    System.out.println("future9" + Thread.currentThread().getName());
//                    Thread.sleep(1000 * 5);
//                    return "akka";
//                }
//            },ec);
//
//            Future<String> fi = f8.zip(f9).map(new Mapper<Tuple2<String, String>, String>() {
//                @Override
//                public String apply(Tuple2<String, String> parameter) {
//                  System.out.println("zip--" + Thread.currentThread().getName());
//                  return parameter._1() + " " + parameter._2();
//                }
//            },ec);
//
//            System.out.println("主线程----" + Thread.currentThread().getName());
//            System.out.println(Await.result(fi, Duration.create(15, "s")));

//异常
//            Future<String> f8 = Futures.future(new Callable<String>() {
//                @Override
//                public String call() throws Exception {
//                    System.out.println("future8" + Thread.currentThread().getName());
//                    Thread.sleep(1000 * 10);
//                    return (1 / 0) + "";
//                }
//            },ec);
//
//            Future<String> f9 = Futures.future(new Callable<String>() {
//                @Override
//                public String call() throws Exception {
//                    System.out.println("future9" + Thread.currentThread().getName());
//                    Thread.sleep(1000 * 5);
//                    return "akka";
//                }
//            },ec);
//
//            Future<String> fi = f8.zip(f9).map(new Mapper<Tuple2<String, String>, String>() {
//                @Override
//                public String apply(Tuple2<String, String> parameter) {
//                  System.out.println("zip--" + Thread.currentThread().getName());
//                  return parameter._1() + " " + parameter._2();
//                }
//            },ec);
//
//            System.out.println("主线程----" + Thread.currentThread().getName());
//            System.out.println(Await.result(fi, Duration.create(15, "s")));
//            recover（对Future的异常进行处理，相当于try..catch中对捕获异常后的处理）
//            Future<Integer> future = Futures.future(new Callable<Integer>() {
//                public Integer call() {
//                    return 1 / 0;
//                }
//            }, ec).recover(new Recover<Integer>() {
//                public Integer recover(Throwable problem) throws Throwable {
//                    System.out.println("捕获到异常：" + problem);
////                if (problem instanceof RuntimeException) {
////                    return 0;
////                } else {
////                    throw problem;
////                }
//                    return -2;    //这里捕获到异常后直接返回新值了，并没有再抛出异常，所以后面的recover不会再收到异常
//                }
//            },ec).recover(new Recover<Integer>() {
//                public Integer recover(Throwable problem) throws Throwable {
//                    System.out.println("捕获到异常：" + problem);
//                    if (problem instanceof ArithmeticException) {  //捕获异常并处理，捕获到后，后面得到的result将会是-1
//                        return -1;
//                    } else {
//                        throw problem;
//                    }
//                }
//            },ec);
//            int result = Await.result(future, Duration.create(1, TimeUnit.SECONDS));
//            System.out.println("result----" + result);
//            recoverWith（和recover很类似，只是捕获到异常后返回Future，使其能够异步并发处理）
//            Future<Integer> future = Futures.future(new Callable<Integer>() {
//                public Integer call() {
//                    return 1 / 0;
//                }
//            }, ec).recoverWith(new Recover<Future<Integer>>() {
//                @Override
//                public Future<Integer> recover(Throwable failure) throws Throwable {
//                    if (failure instanceof ArithmeticException) {
//                        return Futures.future(new Callable<Integer>() {
//                            @Override
//                            public Integer call() throws Exception {
//                                return 0;
//                            }
//                        }, ec);
//                    } else throw failure;
//
//                }
//            },ec);
//            int result = Await.result(future, Duration.create(1, TimeUnit.SECONDS));
//            System.out.println("result----" + result);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
